package com.wyx.springbootspringkafkademo.producer;

import cn.hutool.json.JSONUtil;
import com.wyx.springbootspringkafkademo.entity.Message;
import java.util.concurrent.CompletableFuture;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

/**
 * @author wyx
 * @Date 2025-03-26
 */
@Slf4j
@Component
public class KafkaProducerService {

  public static final String TOPIC = "springboot-spring-kafka-demo-topic";

  @Autowired
  private KafkaTemplate<String, Object> kafkaTemplate;

  public void send(Message message) {
    try {
      CompletableFuture<SendResult<String, Object>> future =
          (CompletableFuture<SendResult<String, Object>>)kafkaTemplate.send(TOPIC, JSONUtil.toJsonStr(message));

      // 异步处理结果
      future.whenComplete((sendResult, ex) -> {
        if (ex != null) {
          // 处理发送失败
          log.error("Failed to send message to topic {}", TOPIC, ex);
        } else {
          // 处理发送成功
          RecordMetadata metadata = sendResult.getRecordMetadata();
          log.info("Message sent successfully to {}-{}-{}", metadata.topic(), metadata.partition(), metadata.offset());
        }
      });
      log.info("send success");
    } catch (Exception e) {
      // 处理异常
      log.error("Failed to send message", e);
    }
  }

  public void sendBatch() {
    for (int i = 0; i < 100; i++) {
      kafkaTemplate.send(TOPIC, JSONUtil.toJsonStr(new Message(i)));
    }
  }
}
